package gupao.concurrency.concurrent.utils;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class DelayQueueSample {

    public static void main(String[] args) {
        Random r = new Random();
        for(int i = 0; i < 10; i++){
            addCache("key_" + i, "value_" + i, r.nextInt(3) + 5);
//            Utils.sleepIgnoreExp(2000);
        }
    }

    static final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>(100000);

    static final DelayQueue<CacheItem> cacheKeyQueueForClean = new DelayQueue<>();

    private static final AtomicBoolean started = new AtomicBoolean(false);

    private static final CacheCleanTask cacheCleanTask = new CacheCleanTask();

    public static boolean addCache(String key, Object val, long expireTimeSecond){
        if(started.compareAndSet(false,true)){
            start();
        }
        CacheItem item = new CacheItem(key, expireTimeSecond);
        if(cacheKeyQueueForClean.offer(item, 200, TimeUnit.MILLISECONDS)){
            cache.putIfAbsent(key,val);
            log.info("add cache, key: {}, val: {}, expireTimeSecond: {}", key, val, expireTimeSecond);
        } else {
            log.error("add cache fail, key: {}, val: {}, expireTimeSecond: {}", key, val, expireTimeSecond);
        }
        return false;
    }

    private static void start(){
        cacheCleanTask.setStop(false);
        new Thread(cacheCleanTask).start();
    }

    public static void stop(){
        if(started.compareAndSet(true,false)){
            cacheCleanTask.setStop(true);
        }
    }

    @Getter
    @Setter
    static class CacheItem implements Delayed{
        private long expireTime; //过期时间，单位纳秒
        private long createdTime; //创建对象的时间戳 精确到纳秒
        private String key;

        public CacheItem(String key, long expireTimeSecond) {
            this.expireTime = TimeUnit.SECONDS.toNanos(expireTimeSecond);
            this.key = key;
            this.createdTime = System.nanoTime();
        }

        /**
         * 该方法返回大于0表示没有过期，否则就过期可以从队列里被获取和删除
         * @param unit
         * @return
         */
        @Override
        public long getDelay(TimeUnit unit) {
            long timePassNanos = System.nanoTime() - createdTime;
            return expireTime - timePassNanos;
        }

        /**
         * 按过期时间先后排序，先过期的排到队首，后过期的排到队尾
         * @param o
         * @return
         */
        @Override
        public int compareTo(Delayed o) {
            long x = getDelay(TimeUnit.NANOSECONDS);
            long y = o.getDelay(TimeUnit.NANOSECONDS);
            return Long.compare(x, y);
        }
    }

    static class CacheCleanTask implements Runnable{

        @Setter
        private volatile boolean stop = false;

        @Override
        public void run() {
            while(!stop){
                try {
                    CacheItem item = cacheKeyQueueForClean.poll(10, TimeUnit.SECONDS);
                    if(item != null){
                        Object removedValue = cache.remove(item.key);
                        log.info("淘汰过期缓存，key = {}, value = {}, 过期时间戳 = {}, 当前时间戳 = {}", item.key, removedValue, item.createdTime + item.expireTime, System.nanoTime());
                    } else {
                        log.debug("No overdue cache item found");
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}
